package com.iussoft.receiveData.service.plugins;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.apache.api.vo.ParamsVo;
import com.apache.api.vo.ResultEntity;
import com.apache.exception.BusinessException;
import com.apache.tools.DateUtils;
import com.apache.tools.StrUtil;
import com.iussoft.receiveData.schedul.SendService;
import com.iussoft.tools.*;
import org.springframework.stereotype.Service;

/**
 * @description: 发送消息到Kafka- 被动推送数据
 * @author: Hou Dayu
 * @date: Created in 2020/3/5
 */
@Service("receive-sendmsg")
public class SendMessagePluginConnector extends SuperPluginconnector {

    private final String EXCHANGE_MODE = "12";

    @Override
    public Object execute(ParamsVo vo) throws Exception {
        ResultEntity result = new ResultEntity();
        String dxpInterfaceInfo = vo.getParamsToStr("dxpInterfaceInfo");
        JSONObject data = JSON.parseObject(dxpInterfaceInfo);
        String topic = getKafkaTopic(1);
        JSONObject json = new JSONObject();
        json.put("dataType","req");
        RequestMessage reqMsg =(RequestMessage)vo.getObj();
        String dataType = "";
        if("2".equals(data.getString("exchangeMode"))){//发送
            dataType = "syncSend";
            topic = getKafkaTopic(2);//异常获取数据
            json.put("dataType","resp");
            json.put("urlParams",reqMsg.getUrlMap() == null ? "" : JSON.toJSONString(reqMsg.getUrlMap()));
            json.put("headParams",JSON.toJSONString(reqMsg.getHeadMap()));
        }
        result.setResult("true");
        result.setMessage("数据接收成功");
        result.setEntity("T");
        JSONObject interfaceInfo = getInterfaceInfo(data.getString("fromServerId"));
        if("1".equals(interfaceInfo.getString("ifTest"))){
            json.put("sourceMsg",interfaceInfo.getString("dataTest"));
        } else {
            if("11".equals(data.getString("exchangeMode")) && "1".equals(interfaceInfo.getString("ifLayOut"))){
                JSONArray arrays = (JSONArray)dataReceiveManager.execute(interfaceInfo,reqMsg,vo.getParamsToStr("traceId"),interfaceInfo.getString("exchangeMode"),topic);
                if(arrays != null && !arrays.isEmpty()){
                    for(int i=0;i<arrays.size();i++) {
                        JSONObject jsobj = arrays.getJSONObject(i);
                        String msg = jsobj.getString("sourceMsg");
                        if(StrUtil.isNull(msg)){
                            continue;
                        }
                        JSONObject jsons = new JSONObject();
                        jsons.put("ifTest",interfaceInfo.getString("ifTest"));
                        if(jsobj.containsKey("mergeParams")){
                            jsons.put("mergeParams",jsobj.get("mergeParams"));
                        }
                        if(jsobj.containsKey("defaultColumn")){//默认列
                            jsons.put("defaultColumn",jsobj.get("defaultColumn"));
                        }
                        jsons.put("sourceMsg",jsobj.getString("sourceMsg").replaceAll("\n","").replaceAll("\r",""));
                        sendMessage(jsons, reqMsg.getInfoByKey("requestPwd"), interfaceInfo, vo,dataType);
                    }
                }
                return result;
            }
            try {
                if(EXCHANGE_MODE.equals(data.getString("exchangeMode"))) {//异步获取数据
                    String resultData = sendDataMsg(data.getString("exchangeMode"), json, interfaceInfo, reqMsg);
                    if (StrUtil.isNull(resultData)) {
                        throw new BusinessException("接口数据返回空");
                    }
                    json.put("sourceMsg",resultData.replaceAll("\n","").replaceAll("\r",""));
                } else {
                    json.put("sourceMsg",reqMsg.getSourceMsg().replaceAll("\n","").replaceAll("\r",""));
                }
            }catch (BusinessException e){
                result.setMessage("获取" + interfaceInfo.getString("interfaceUri") + "接口失败，错误信息为: " + e.getMessage());
                LoggerUtil.infoSource("exception = {}; infoType = exception; traceId = {}; interfaceCode = {}; requestTime = {}; errorMessage = {}; requestIp = {}; exeSys = receive; level = {}",e.getClass().getSimpleName(), vo.getParamsToStr("traceId"), reqMsg.getInfoByKey("fromInterfaceCode"), DateUtils.Now.fmt_yyyyMMdd_HHmmss(), result.getMessage(),vo.getParamsToStr("requestUseIp"),"ERROR");
                return result;
            }
        }
        json.put("ifTest",interfaceInfo.getString("ifTest"));
        /*json.put("sendTime",System.currentTimeMillis());
        json.put("interfaceCode",reqMsg.getInfoByKey("fromInterfaceCode"));
        json.put("infterfaceId",reqMsg.getInfoByKey("fromServerId"));
        json.put("infterfacePwd",reqMsg.getInfoByKey("requestPwd"));
        json.put("sysEname",reqMsg.getInfoByKey("sysEname"));
        json.put("traceId",vo.getParamsToStr("traceId"));
        json.put("requestUseIp",vo.getParamsToStr("requestUseIp"));
        json.put("requestCode",vo.getParamsToStr("requestCode"));
        if(kafkaSimpleSender.getKafkaEnabled()) {
            kafkaSimpleSender.sendMsg(topic,"0", json.toJSONString());
        }else {
            SendService sendService = new SendService(json,dataType);
            dynamicTask.submitRun(sendService);
        }*/
        sendMessage(json, reqMsg.getInfoByKey("requestPwd"), interfaceInfo, vo, dataType);
        return result;
    }

    /**
     * 同步数据
     */
    private void sendMessage(JSONObject json,String requestPwd , JSONObject interfaceInfo, ParamsVo vo,String dataType){
        json.put("sendTime", System.currentTimeMillis());
        json.put("interfaceCode", interfaceInfo.getString("interfaceCode"));
        json.put("infterfaceId", interfaceInfo.getString("infoId"));
        json.put("sysEname", interfaceInfo.getString("sysEname"));
        json.put("infterfacePwd", requestPwd);
        json.put("traceId", vo.getParamsToStr("traceId"));
        json.put("requestUseIp", vo.getParamsToStr("requestUseIp"));
        json.put("requestCode", vo.getParamsToStr("requestCode"));
        if(kafkaSimpleSender.getKafkaEnabled()) {
            String topic = getKafkaTopic(1);
            String key ="0";
            kafkaSimpleSender.sendMsg(topic, key, json.toJSONString());
        } else {
            dynamicTask.submitRun(new SendService(json, dataType));
        }
    }
}
